package com.lagou.rpc.consumer.proxy;

import com.alibaba.fastjson.JSON;
import com.lagou.rpc.common.RpcRequest;
import com.lagou.rpc.common.RpcResponse;
import com.lagou.rpc.consumer.client.RpcClient;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.springframework.util.CollectionUtils;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @description: 客户端代理类-创建代理对象
 * 1.封装request请求对象
 * 2.创建RpcClient对象
 * 3.发送消息
 * 4.返回结果
 * @author: yx
 * @date: 2022/1/11 16:31
 */
public class RpcClientProxy {

    private static Map<String, RpcClient> clientMap = new ConcurrentHashMap<>();

    //定时线程池
    private static ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);

    public static Object createProxy(Class serviceClass, CuratorFramework zkClient) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{serviceClass},
                new InvocationHandler() {
                    @Override
                    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                        //1.封装request请求对象
                        RpcRequest rpcRequest = new RpcRequest();
                        rpcRequest.setRequestId(UUID.randomUUID().toString());
                        rpcRequest.setClassName(method.getDeclaringClass().getName());
                        rpcRequest.setMethodName(method.getName());
                        rpcRequest.setParameters(args);
                        rpcRequest.setParameterTypes(method.getParameterTypes());
                        //2。获取所有节点列表
                        String parentPath = "/zk";
                        List<String> nodeList = zkClient.getChildren().forPath(parentPath);
                        if (CollectionUtils.isEmpty(nodeList)) {
                            throw new RuntimeException("连接失败，获取不到节点列表");
                        }

                        //获取连接对象
                        RpcClient rpcClient = getRpcClient(parentPath, nodeList, zkClient);

                        try {
                            //4.发送消息
                            long start = System.currentTimeMillis();
                            System.out.println(String.format("给ip: %s,port: %s发送消息", rpcClient.getIp(), rpcClient.getPort()));
                            Object responseMsg = rpcClient.send(JSON.toJSONString(rpcRequest));
                            RpcResponse rpcResponse = JSON.parseObject(responseMsg.toString(), RpcResponse.class);
                            if (rpcResponse.getError() != null) {
                                throw new RuntimeException(rpcResponse.getError());
                            }
                            //5.返回结果
                            Object result = rpcResponse.getResult();
                            Object resultJson = JSON.parseObject(result.toString(), method.getReturnType());
                            long end = System.currentTimeMillis();
                            long time = end - start;
                            //往节点上记录耗时和系统时间
                            String timeData = time + ":" + end;
                            zkClient.setData().forPath(rpcClient.getNodePath() + "/time", timeData.getBytes(StandardCharsets.UTF_8));
                            System.out.println("记录耗时数据：" + timeData);
                            //返回结果
                            return resultJson;
                        } catch (Exception e) {
                            System.out.println("出异常了: " + e.getMessage());
                            throw e;
                        }
                    }
                });
    }

    /**
     * 获取客户端连接
     *
     * @param parentPath 父级路径
     * @param nodeList   子节点列表
     * @param zkClient   zk客户端
     * @return
     * @throws Exception
     */
    private static RpcClient getRpcClient(String parentPath, List<String> nodeList, CuratorFramework zkClient) throws Exception {
        //最小的响应时间
        Long minTime = -1L;
        //最小响应时间对应的连接
        RpcClient minRpcClient = null;

        for (int i = 0; i < nodeList.size(); i++) {
            String s = nodeList.get(i);
            //节点路径
            String nodePath = parentPath + "/" + s;
            //判断节点的状态是否在线
            Stat stat = zkClient.checkExists().forPath(nodePath + "/stat");
            if (stat == null) {
                //不在线直接跳过
                return null;
            }
            //获取节点里的服务端数据
            byte[] bytes = zkClient.getData().forPath(nodePath);
            String data = new String(bytes, StandardCharsets.UTF_8);
            String[] split = data.split(":");
            String ip = split[0];
            String port = split[1];
            RpcClient rpcClient = clientMap.get(nodePath);

            String timePath = nodePath + "/time";

            if (rpcClient == null) {
                rpcClient = new RpcClient(ip, Integer.valueOf(port));
                clientMap.put(nodePath, rpcClient);

                //定时任务
                scheduledThreadPool.scheduleAtFixedRate(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            Stat timeStat = zkClient.checkExists().forPath(timePath);
                            if (timeStat != null) {
                                byte[] timeData = zkClient.getData().forPath(timePath);

                                if (timeData != null && timeData.length > 0) {
                                    String time = new String(timeData, StandardCharsets.UTF_8);
                                    long currentTimeMillis = System.currentTimeMillis();
                                    String[] split1 = time.split(":");
                                    if ((currentTimeMillis - Long.parseLong(split1[1])) > 20 * 1000) {
                                        //超过5秒清除数据
                                        System.out.println("定时任务开始，数据为：" + time + " timePath:" + timePath);
                                        System.out.println("超过5秒清除数据");
                                        zkClient.setData().forPath(timePath, null);
                                    }
                                }
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }, 5, 20, TimeUnit.SECONDS);
            }

            rpcClient.setNodePath(nodePath);
            //获取时间并进行比较
            Stat timeStat = zkClient.checkExists().forPath(timePath);
            byte[] timeData;
            if (timeStat == null) {
                timeData = new byte[]{};
                zkClient.create().withMode(CreateMode.EPHEMERAL).forPath(timePath, null);
            } else {
                timeData = zkClient.getData().forPath(timePath);
            }
            if (timeData != null && timeData.length > 0) {
                String timeDataStr = new String(timeData, StandardCharsets.UTF_8);
                String[] split1 = timeDataStr.split(":");
                long l = Long.parseLong(split1[0]);
                if ((minTime != null && l < minTime) || i == 0) {
                    //如果响应时间要小于当前最小响应时间
                    minTime = l;
                    minRpcClient = rpcClient;
                }
            } else {
                minTime = null;
                minRpcClient = rpcClient;
            }
            //对状态节点进行监听 dataIsCompressed:是否进行数据压缩
            PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, nodePath, false);
            pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
            pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                    if (event.getData().getPath().equals(nodePath + "/stat") && event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED) {
                        Thread.sleep(1000);
                        RpcClient rpcClient = new RpcClient(ip, Integer.parseInt(port));
                        clientMap.put(nodePath, rpcClient);
                        System.out.println(data + "重新建立连接成功");
                    }
                    if (event.getData().getPath().equals(nodePath + "/stat") && event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
                        RpcClient rpcClient1 = clientMap.get(nodePath);
                        if (rpcClient1 != null) {
                            rpcClient1.close();
                        }
                        clientMap.remove(nodePath);
                        System.out.println("删除节点: " + data + "监听成功");
                    }
                }
            });
        }
        return minRpcClient;
    }
}
